package cn.doitedu.rtdw.data_etl;

import cn.doitedu.rtdw.udf.Gps2Geo;
import cn.doitedu.rtdw.udf.Map2JsonStr;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author: deep as the sea
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2023/4/4
 * @Desc: 学大数据，到多易教育
 *  用户行为事件日志数据公共维度退维处理任务
 *
 *  本任务的核心要点：
 *     1. 这个任务是干嘛的：公共维度退维
 *     2. 怎么做的？ 各种维表用各种技术手段同步到hbase中，然后在此任务中通过lookup关联来查询各维度字段，加入到日志数据中
 *     3. 有没有什么问题？
 *         -- lookup查询可能会对hbase造成严重的查询压力，可以通过开启lookup缓存来缓解；但是对user维表不适用，因为user维表中的数据有可能会更新
 *         -- lookup查询可能因为目标维表数据延迟到达而查询失败，可以用flink-1.16.x的新特性语法开启延迟重试机制；如果是老版本，可以自己写api利用定时器来达到同样效果
 *
 **/
public class Etl01_UserEventsOds2Dwd {
    public static void main(String[] args) {

        // 创建编程入口
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(2000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:/d:/ckpt");
        env.getCheckpointConfig().setTolerableCheckpointFailureNumber(3);
        env.setParallelism(1);
        // 状态后端
        env.setStateBackend(new HashMapStateBackend());
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        // 创建kafka连接器表，映射kafka中的行为日志数据 topic: ods-events
        /* *
         * flinksql中创建 连接器映射表，在schema定义中可以声明如下内容：
         *    物理字段
         *    表达式字段【 利用表达式字段还可以声明 处理时间语义字段 】
         *    元数据字段
         *    watermark
         */
        tenv.executeSql(
                " CREATE TABLE events_source (                               "
                        // 物理字段
                        + "     username     string,                            "
                        + "     session_id   string,                            "
                        + "     eventId      string,                            "
                        + "     actionTime   bigint,                            "
                        + "     lat          double,                            "
                        + "     lng          double,                            "
                        + "     release_channel   string,                       "
                        + "     device_type       string,                       "
                        + "     properties   map<string,string>,                "
                        // 表达式字段,声明了 process time语义的时间戳字段，用于后续的lookup join
                        + "     proc_time   AS  proctime(),                    "
                        + "     event_time  AS  to_timestamp_ltz(actionTime,3),"
                        // 声明watermark
                        + "     watermark for event_time as event_time - interval '0' second "
                        + " ) WITH (                                            "
                        + "  'connector' = 'kafka',                             "
                        + "  'topic' = 'ods-events',                            "
                        + "  'properties.bootstrap.servers' = 'doitedu:9092',   "
                        + "  'properties.group.id' = 'goo1',                    "
                        + "  'scan.startup.mode' = 'latest-offset',             "
                        + "  'value.format'='json',                             "
                        + "  'value.json.fail-on-missing-field'='false',        "
                        + "  'value.fields-include' = 'EXCEPT_KEY'              "
                        + " )                                                   ");

        // 创建hbase连接器表，映射 hbase中的维表： dim_user_info
        tenv.executeSql("CREATE TABLE user_hbase( " +
                " username STRING, " +
                " f ROW<id BIGINT,phone STRING, status INT, create_time TIMESTAMP(3), gender INT, birthday DATE, province STRING, city STRING, job STRING, source_type INT>, " +
                " PRIMARY KEY (username) NOT ENFORCED " +
                ") WITH (                             " +
                " 'connector' = 'hbase-2.2',          " +
                " 'table-name' = 'dim_user_info',     " +
                " 'zookeeper.quorum' = 'doitedu:2181' " +
                ")");

        // 创建hbase连接器表，映射 hbase中的维表： dim_geo_area
        tenv.executeSql("CREATE TABLE geo_hbase( " +
                " geohash STRING, " +
                " f ROW<p STRING,c STRING,r STRING>, " +
                " PRIMARY KEY (geohash) NOT ENFORCED " +
                ") WITH (                             " +
                " 'connector' = 'hbase-2.2',          " +
                " 'table-name' = 'dim_geo_area',     " +
                " 'zookeeper.quorum' = 'doitedu:2181' " +
                ")");


        // 创建hbase连接器表，映射 hbase中的维表： dim_page_info
        tenv.executeSql("CREATE TABLE page_hbase( " +
                " url_prefix STRING, " +
                " f ROW<pt STRING,sv STRING>, " +
                " PRIMARY KEY (url_prefix) NOT ENFORCED " +
                ") WITH (                             " +
                " 'connector' = 'hbase-2.2',          " +
                " 'table-name' = 'dim_page_info',     " +
                " 'zookeeper.quorum' = 'doitedu:2181' " +
                ")");

        // 编写运算代码（lookup关联）
        /* *
         * 要点1： lookup join的特别语法  LEFT JOIN 维表 FOR SYSTEM_TIME AS OF 主表.处理时间语义字段
         * 要点2： 从hbase映射表中取family里面的字段： f.字段名
         * 要点3： flink-1.16.x的新特性，在lookup的sql语法中支持控制延迟重试（利用hint传入重试相关参数）
         */
        tenv.createTemporaryFunction("geo", Gps2Geo.class);
        tenv.executeSql(
                " CREATE TEMPORARY VIEW wide_view AS                                                                                                              "
                        // flink-1.16.0增强的新特性：lookup查询支持延迟重试
                        +" SELECT  /*+ LOOKUP('table'='user_hbase', 'retry-predicate'='lookup_miss', 'retry-strategy'='fixed_delay', 'fixed-delay'='5s','max-attempts'='3') */"
                        +" u.f.id as user_id,                                                                                                                              "
                        +" e.username,                                                                                                                                     "
                        +" e.session_id,                                                                                                                                   "
                        +" e.eventId as event_id,                                                                                                                          "
                        +" e.actionTime as action_time,                                                                                                                    "
                        +" e.lat,e.lng,                                                                                                                                    "
                        +" e.release_channel,                                                                                                                              "
                        +" e.device_type,                                                                                                                                  "
                        +" e.properties,                                                                                                                                   "
                        +" u.f.phone as register_phone,                                                                                                                    "
                        +" u.f.status as user_status,                                                                                                                      "
                        +" u.f.create_time as register_time,                                                                                                               "
                        +" u.f.gender as register_gender,                                                                                                                  "
                        +" u.f.birthday as register_birthday,                                                                                                              "
                        +" u.f.province as register_province,                                                                                                              "
                        +" u.f.city as register_city,                                                                                                                      "
                        +" u.f.job as register_job,                                                                                                                        "
                        +" u.f.source_type as register_source_type,                                                                                                        "
                        +" g.f.p as gps_province ,                                                                                                                         "
                        +" g.f.c as gps_city,                                                                                                                              "
                        +" g.f.r as gps_region,                                                                                                                            "
                        +" p.f.pt as page_type,                                                                                                                            "
                        +" p.f.sv as page_service                                                                                                                          "
                        +" FROM events_source AS e                                                                                                                         "
                        +" LEFT JOIN user_hbase FOR SYSTEM_TIME AS OF e.proc_time AS u  ON e.username = u.username                                                         "
                        // geohash，反转关联hbase中的rowkey（因为hbase中就是反转存储的，避免热点问题）
                        +" LEFT JOIN geo_hbase FOR SYSTEM_TIME AS OF e.proc_time AS g ON REVERSE(geo(e.lat,e.lng)) = g.geohash                                             "
                        +" LEFT JOIN page_hbase FOR SYSTEM_TIME AS OF e.proc_time AS p ON regexp_extract(e.properties['url'],'(^.*/).*?') = p.url_prefix                   "
        );


        // 创建kafka连接器表，映射kafka中的 dwd层的行为数据 topic : dwd-events-detail
        tenv.executeSql(
                "  CREATE TABLE dwd_kafka(                           "
                        +"     user_id           BIGINT,                     "
                        +"     username          string,                     "
                        +"     session_id        string,                     "
                        +"     event_Id          string,                     "
                        +"     event_time        bigint,                     "
                        +"     lat               double,                     "
                        +"     lng               double,                     "
                        +"     release_channel   string,                     "
                        +"     device_type       string,                     "
                        +"     properties        map<string,string>,         "
                        +"     register_phone    STRING,                     "
                        +"     user_status       INT,                        "
                        +"     register_time     TIMESTAMP(3),               "
                        +"     register_gender   INT,                        "
                        +"     register_birthday DATE,                       "
                        +"     register_province STRING,                     "
                        +"     register_city        STRING,                  "
                        +"     register_job         STRING,                  "
                        +"     register_source_type INT,                     "
                        +"     gps_province STRING,                          "
                        +"     gps_city     STRING,                          "
                        +"     gps_region   STRING,                          "
                        +"     page_type    STRING,                          "
                        +"  page_service STRING                              "
                        +" ) WITH (                                          "
                        +"  'connector' = 'kafka',                           "
                        +"  'topic' = 'dwd-events-detail',                   "
                        +"  'properties.bootstrap.servers' = 'doitedu:9092', "
                        +"  'properties.group.id' = 'testGroup',             "
                        +"  'scan.startup.mode' = 'earliest-offset',         "
                        +"  'value.format'='json',                           "
                        +"  'value.json.fail-on-missing-field'='false',      "
                        +"  'value.fields-include' = 'EXCEPT_KEY')           ");

        // 将关联好的宽表数据插入kafka
        tenv.executeSql("insert into dwd_kafka select * from wide_view");



        // 创建doris连接器表，映射doris中的dwd层的行为数据明细宽表： dwd.user_events_detail
        /* *
         *  这里面有两个细节：
         *    1. 上面退维的结果中有properties字段，且其数据类型为Map，而doris中并不支持map类型，所以我们把properties这个map数据转成json字符串，写入doris
         *    2. 上面退维的结果中没有dt字段，而doris的目标表必须要有一个dt来做分区，我们可以从退维结果的actionTime字段中用函数得到dt
         */
        tenv.executeSql(
                " CREATE TABLE dwd_doris  (         "
                        + "     gps_province         VARCHAR(16),   "
                        + "     gps_city             VARCHAR(16),   "
                        + "     gps_region           VARCHAR(16),   "
                        + "     dt                   DATE,          "
                        + "     user_id              BIGINT,        "
                        + "     username             VARCHAR(20),   "
                        + "     session_id           VARCHAR(20),   "
                        + "     event_id             VARCHAR(10),   "
                        + "     event_time           bigint,        "
                        + "     lat                  DOUBLE,        "
                        + "     lng                  DOUBLE,        "
                        + "     release_channel      VARCHAR(20),   "
                        + "     device_type          VARCHAR(20),   "
                        + "     properties           VARCHAR(40),   "
                        + "     register_phone       VARCHAR(20),   "
                        + "     user_status          INT,           "
                        + "     register_time        TIMESTAMP(3),  "
                        + "     register_gender      INT,           "
                        + "     register_birthday    DATE,          "
                        + "     register_province    VARCHAR(20),   "
                        + "     register_city        VARCHAR(20),   "
                        + "     register_job         VARCHAR(20),   "
                        + "     register_source_type INT        ,   "
                        + "     page_type            VARCHAR(20),   "
                        + "     page_service         VARCHAR(20)    "
                        + " ) WITH (                               "
                        + "    'connector' = 'doris',              "
                        + "    'fenodes' = 'doitedu:8030',         "
                        + "    'table.identifier' = 'dwd.user_events_detail',  "
                        + "    'username' = 'root',                "
                        + "    'password' = '',                    "
                        + "    'sink.label-prefix' = 'doris_label" + System.currentTimeMillis() + "'"
                        + " )                                         "
        );

        // 将退维结果 略微加工后，写入doris目标映射表
        /* *
         * 加工的要点：
         *       根据目标doris表，调整字段顺序
         *       根据行为事件生成 DT 字段
         *       将properties转成json字符串形式
         */
        tenv.createTemporaryFunction("toJson", Map2JsonStr.class);
        tenv.executeSql("INSERT INTO dwd_doris                                     "
                + " SELECT                                                                         "
                + "     gps_province         ,                                                     "
                + "     gps_city             ,                                                     "
                + "     gps_region           ,                                                     "
                + "     TO_DATE(DATE_FORMAT(TO_TIMESTAMP_LTZ(action_time, 3),'yyyy-MM-dd')) as dt,  "
                + "     user_id              ,                                                     "
                + "     username             ,                                                     "
                + "     session_id           ,                                                     "
                + "     event_id             ,                                                     "
                + "     action_time           ,                                                     "
                + "     lat                  ,                                                     "
                + "     lng                  ,                                                     "
                + "     release_channel      ,                                                     "
                + "     device_type          ,                                                     "
                + "     toJson(properties) as properties     ,                                     "
                + "     register_phone       ,                                                     "
                + "     user_status          ,                                                     "
                + "     register_time        ,                                                     "
                + "     register_gender      ,                                                     "
                + "     register_birthday    ,                                                     "
                + "     register_province    ,                                                     "
                + "     register_city        ,                                                     "
                + "     register_job         ,                                                     "
                + "     register_source_type ,                                                     "
                + "     page_type            ,                                                     "
                + "     page_service                                                               "
                + " FROM   wide_view                                                               "
        );
    }
}
